package com.wms.provider.controller;

import com.wms.provider.util.MessageSplitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Description: 生产者发送消息
 *
 * @version 1.0
 * @author: wangqiang
 * @date: 2023/3/12 23:41
 */
@Slf4j
@RestController("/provider")
public class ProviderController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 主题名
    private static final String topic = "test_topic";

    /**
     * 测试
     */
    @GetMapping("/send")
    public void send() {
        String text = "吕布-貂蝉";
        Message<String> message = MessageBuilder.withPayload(text).build();
        log.info("开始发送...");
        // destination 参数, 可以由:分开，第一个参数为topic，第二个参数为tag，默认可以不设置tag参数，tag的设置应该根据实际使用场景来绝对。
        rocketMQTemplate.send(topic, message);
        log.info("已发送...");
    }

    /**
     * TODO 一、基本消息
     */


    /**
     * 1.同步消息：这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
     */
    @GetMapping("/syncSend")
    public void syncSend() {
        String text = "基本信息案例-同步发送" + System.currentTimeMillis();
        rocketMQTemplate.syncSend("base_topic", text);
        log.info("同步发送成功...");
    }

    /**
     * 2.异步消息：用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
     */
    @GetMapping("/asyncSend")
    public void asyncSend() {
        String text = "基本信息案例-异步发送" + System.currentTimeMillis();
        rocketMQTemplate.asyncSend("base_topic", text, new SendCallback() {

            // SendCallback接收异步返回结果的回调
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送 - 发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("异步发送 - 发送失败");
                throwable.printStackTrace();
            }
        });
    }

    /**
     * 3.单向发送消息：这种方式主要用在不关心发送结果的场景，例如日志发送。
     */
    @GetMapping("/sendOneWay")
    public void sendOneWay() {
        String text = "基本信息案例-单向发送" + System.currentTimeMillis();
        rocketMQTemplate.sendOneWay("base_topic", text);
        log.info("单向发送成功...");
    }


    /**
     * TODO 二、顺序消息
     */
    @GetMapping("/syncSendOrderly")
    public void syncSendOrderly() {
        for (int i = 1; i <= 10; i++) {

            // 设置一个延时,表示同一个消息先后进入到队形中
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String text = "顺序消息ID: " + i;
            log.info("{}...正在发送", text);
            // 第一个参数为topic，第二个参数为内容，第三个参数为Hash值，不同hash值在不同的队列中
            rocketMQTemplate.syncSendOrderly("order_topic", text, "order");
        }
        log.info("发送成功...");
    }

    /**
     * TODO 三、延时消息
     */
    @GetMapping("/scheduled")
    public void scheduled() {

        String text = "延时消息" + System.currentTimeMillis();
        log.info("{}...正在发送", text);
        // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        Message<String> message = MessageBuilder.withPayload(text).build();
        // 设置延时等级2,这个消息将在5s之后发送
        rocketMQTemplate.syncSend("scheduled_topic", message, 1000, 2);
        log.info("延时消息发送成功...");
    }

    /**
     * TODO 四、标签过滤消息 (一个应用尽可能用一个Topic，消息子类型用tag来标识，tag可以由应用自由设置。)
     */
    @GetMapping("/tag")
    public void tag() {

        String text = "标签过滤消息," + System.currentTimeMillis();
        log.info("{}...正在发送", text);
        // 任何类型的send方法均可以指定TAG，默认可以不指定则为*
        Message<String> message = MessageBuilder.withPayload(text).build();
        // 主题:标签
        rocketMQTemplate.syncSend("tag_topic:TAG-A", message);
        log.info("标签过滤消息发送成功...");
    }

    /**
     * TODO 五、SQL92过滤消息
     */
    @GetMapping("/convertAndSend")
    public void convertAndSend() {

        String text = "SQL92过滤消息," + System.currentTimeMillis();
        log.info("{}...正在发送", text);
        Message<String> message = MessageBuilder.withPayload(text).build();
        Map<String, Object> map = new HashMap<>(4);
        map.put("a", 2);
        map.put("b", 10);
        rocketMQTemplate.convertAndSend("sql_topic", message, map);
        log.info("SQL92过滤消息发送成功...");
    }

    /**
     * TODO 六、批量消息
     */
    @GetMapping("/batchSend")
    public void batchSend() {

        String text = "批量消息";
        List<Message> messageList = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            messageList.add(MessageBuilder.withPayload(text + "--" + i).build());
        }
        log.info("{}...正在发送", text);

        // 大小超过4M,分割消息
        MessageSplitter splitter = new MessageSplitter(messageList);
        while (splitter.hasNext()) {
            List<Message> next = splitter.next();
            SendResult result = rocketMQTemplate.syncSend("batch_topic", next);
            if (result.getSendStatus() == SendStatus.SEND_OK) {
                log.info("发送批量消息成功!消息ID为:{}", result.getMsgId());
            } else {
                log.info("发送批量消息失败!消息ID为:{},消息状态为:{}", result.getMsgId(), result.getSendStatus());
            }
        }
        log.info("批量消息发送成功...");
    }


    /**
     * TODO 七、回馈消息
     */
    @GetMapping("/reply")
    public void reply() {

        // 如果消费者没有回馈消息，则不会发送下一条消息
        // 同步发送消息并且返回一个String类型的结果。
        for (int i = 1; i <= 10; i++) {
            String text = "回馈消息" + "--" + i;
            log.info("发送" + text);
            String replyString = rocketMQTemplate.sendAndReceive("reply_topic", text, String.class);
            log.info("消费者返回的消息：" + replyString);
        }
    }
}
